package com.zyh.flink.day07.evictor;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.delta.DeltaFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.DeltaEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class DeltaEvictorTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<Double> dataStream = environment.socketTextStream("hadoop10", 9998)
                .map(new MapFunction<String, Double>() {
                    @Override
                    public Double map(String value) throws Exception {
                        return Double.valueOf(value);
                    }
                });

        AllWindowedStream<Double, TimeWindow> windowedStream = dataStream
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .evictor(DeltaEvictor.of(2.0, new DeltaFunction<Double>() {
                    @Override
                    public double getDelta(Double oldDataPoint, Double newDataPoint) {
                        System.out.print("oldDataPoint = " + oldDataPoint);
                        System.out.println(",newDataPoint = " + newDataPoint);
                        return Math.abs(oldDataPoint - newDataPoint);//将和最新一个元素相差超过2.0的旧元素剔除掉
                    }
                }));

        SingleOutputStreamOperator<String> result = windowedStream
                .process(new ProcessAllWindowFunction<Double, String, TimeWindow>() {
            @Override
            public void process(Context context, Iterable<Double> elements, Collector<String> out) throws Exception {
                out.collect(elements.toString());
            }
        });

        result.print();

        environment.execute("deltaEvictor");
    }
}
